package com.neo.push.server.connection;

import com.neo.push.core.*;
import com.neo.push.core.connection.SocketConnection;
import com.neo.push.server.SafetyMessageTask;
import com.neo.push.server.message.PushMessageMonitor;
import io.netty.util.Timeout;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;

/**
 * Created by lock on 17-8-2.
 */
public class BasePushMessageHandler implements PushMessageHandler {

    private PushMessageMonitor messageMonitor;
    private Logger logger = LoggerFactory.getLogger(PushMessageHandler.class);

    public BasePushMessageHandler() {

    }

    public BasePushMessageHandler(PushMessageMonitor pushMessageMonitor){
        this.messageMonitor = pushMessageMonitor;
    }

    /**
     * 保证当前消息必达，如果不到，则抛出异常
     * @param pushTextMessage   推送的消息
     */
    private void holdMessage(SocketConnection connection,SafetySocketMessage pushTextMessage){
        if(!(connection instanceof SafetyMessageConnection)){
            //当前连接不支持保持消息的操作
            logger.error(" 不支持当前连接 !");
            connection.close();
            return;
        }
        SafetyMessageConnection safetyConnection = (SafetyMessageConnection) connection;
        if(safetyConnection.isHoleMessage()){
            String messageId = pushTextMessage.getMessage().getMessageId();
            messageMonitor.addTask(new MessageTimeoutTask(messageId,safetyConnection), safetyConnection.timeout(),safetyConnection.timeoutUnit());
        }
    }



    /**
     * 如何处理一个消息回执
     * @param receipt       消息回执
     */
    private void handlePushReceipt(Receipt receipt) throws Exception {
        Timeout timeout = messageMonitor.removeTimeout(receipt.getMessageId());
        if(null!=timeout){
            timeout.cancel();
        }
        messageMonitor.receipt(receipt.getMessageId());
    }



    @Override
    public void acceptedMessage(SocketConnection connection, TextMessage message) {
        try {
            handlePushReceipt((Receipt) message);
        } catch (Exception e) {
            e.printStackTrace();
            connection.close();
        }
    }


    @Override
    public boolean isSupported(TextMessage message) {
        return (message instanceof Receipt);
    }

    /**
     * 发送消息
     * @param connection    当前连接所拥有的channel
     * @param textMessage   文本消息
     * @throws Exception
     */
    @Override
    public void writeMessage(SocketConnection connection, SafetySocketMessage textMessage) throws Exception {
        holdMessage(connection,textMessage);
        connection.ctx().writeAndFlush(textMessage);
    }

    @Override
    public List<AcceptPush> loadOfflinePush(RegisterMessage registerMessage) throws Exception {
        return messageMonitor.loadOfflineMessage(registerMessage.getTargetName());
    }


    @Override
    public void handleMessageTimeout(String messageId) throws Exception {
        messageMonitor.handleMessageTimeout(messageId);
    }


    /**
     * 能执行到当前的任务，并且任务没有取消的情况下，都认为已经超时
     */
    private class MessageTimeoutTask implements SafetyMessageTask {

        private final String messageId;
        private final SafetyMessageConnection connection;

        MessageTimeoutTask(String messageId,SafetyMessageConnection connection){
            this.messageId = messageId;
            this.connection = connection;
        }

        @Override
        public void run(Timeout timeout) throws Exception {
            //移除当前timeout
            messageMonitor.removeTimeout(messageId);
            //如果这个信息是取消过了的
            logger.info(" timeout is "+timeout.isCancelled());
            if(timeout.isCancelled()) return;
            //处理消息超时
            connection.handleMessageTimeout(messageId);
        }

        @Override
        public String messageId() {
            return messageId;
        }
    }


    public void setMessageMonitor(PushMessageMonitor messageMonitor) {
        this.messageMonitor = messageMonitor;
    }

}
